package cn.com.cennavi.service;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;

import org.apache.spark.api.java.function.PairFlatMapFunction;

import cn.com.cennavi.bean.HighPrecisionCombineDirTraffic;
import cn.com.cennavi.bean.HighPrecisionSingleDirTraffic;
import scala.Tuple2;

/**
 * 分岔口道路合并
 * 
 * @ClassName: RoadTrafficCombine
 * @Description: TODO（这里用一句话描述这个类的作用）
 * @author bull
 * @date 2016年4月13日 下午1:08:08
 *
 */
public class RoadTrafficCombine implements PairFlatMapFunction<Tuple2<String, Iterable<HighPrecisionSingleDirTraffic>>, String, HighPrecisionCombineDirTraffic> {
	private static final long serialVersionUID = 1L;

	@Override
	public Iterator<Tuple2<String, HighPrecisionCombineDirTraffic>> call(Tuple2<String, Iterable<HighPrecisionSingleDirTraffic>> arg0) throws Exception {
		// TODO Auto-generated method stub
		ArrayList<Tuple2<String, HighPrecisionCombineDirTraffic>> result = new ArrayList<>();
		String difPointId_straightAngle_roadClass = arg0._1;
		Iterable<HighPrecisionSingleDirTraffic> traffics = arg0._2;
		String[] strs = difPointId_straightAngle_roadClass.split("\\_");
		String pointId = strs[0];
		int roadClass = Integer.parseInt(strs[2]);
		if (roadClass == 0) {
			// 封闭道路
			try {
				// 最多包含了2个方向的数据，直行和左入或左出或右入或右出
				getTrafficCloseRoad(result, traffics, pointId, roadClass);
			} catch (Exception e) {
				// TODO: handle exception
				e.printStackTrace();
			}
		}
		if (roadClass == 1) {
			// 地面道路，最多包含了3个方向的数据，直行和左转或右转
			HighPrecisionCombineDirTraffic ss = new HighPrecisionCombineDirTraffic();
			result.add(new Tuple2<String, HighPrecisionCombineDirTraffic>(pointId, ss));
		}

		return result.iterator();
	}

	private void getTrafficCloseRoad(ArrayList<Tuple2<String, HighPrecisionCombineDirTraffic>> result, Iterable<HighPrecisionSingleDirTraffic> traffics, String pointId, int roadClass) {
		// TODO Auto-generated method stub
		int num = 0;
		ArrayList<HighPrecisionSingleDirTraffic> trafficDataList = new ArrayList<>();

		for (HighPrecisionSingleDirTraffic traffic : traffics) {
			num++;
			trafficDataList.add(traffic);
			// System.out.println("combinereslt1:list:" + traffic.toString());

		}
		// 直行弯路只有一条的情况，需要补全另一条
		if (num == 1) {
			HighPrecisionCombineDirTraffic outLine = getOutLine(trafficDataList.get(0), roadClass);
			result.add(new Tuple2<String, HighPrecisionCombineDirTraffic>(pointId, outLine));
		} else {
			if (num >= 2) {
				System.out.println("有" + num + "条方向数据进行合并!");
			}
			ArrayList<HighPrecisionCombineDirTraffic> dataList = new ArrayList<>();
			for (HighPrecisionSingleDirTraffic traffic : trafficDataList) {
				HighPrecisionCombineDirTraffic outLine = getOutLine(traffic, roadClass);
				dataList.add(outLine);
			}
			HighPrecisionCombineDirTraffic outLine = combineTwoLine(dataList);
			result.add(new Tuple2<String, HighPrecisionCombineDirTraffic>(pointId, outLine));

			if (num >= 2) {
				System.out.println("合并后数据：" + outLine.toString1());
			}
		}
	}

	private HighPrecisionCombineDirTraffic getOutLine(HighPrecisionSingleDirTraffic highPrecisionSingleDirTraffic, int roadClass) {
		// TODO Auto-generated method stub
		String[] arr = highPrecisionSingleDirTraffic.toString().split(",");
		String id = arr[0];
		String lonlat = arr[1];
		String angle = arr[2];
		String dir = arr[3];
		String flag = arr[4];
		String trajMakeTime = "0000";
		String bef_dis_speed = arr[6];
		String aft_dis_speed = arr[7];
		int total_len = Integer.parseInt(arr[8]);
		int bef_len = Integer.parseInt(arr[9]);
		int aft_len = Integer.parseInt(arr[10]);
		String carIdTimeSeq = arr[11];
		String[] bef_dis_speed_other = bef_dis_speed.split("\\|");
		StringBuffer bef_sb = new StringBuffer();
		for (String dis_speed : bef_dis_speed_other) {
			String[] disspeed = dis_speed.split("\\_");
			bef_sb.append(disspeed[0] + "_" + "-1" + "|");
		}
		String[] aft_dis_speed_other = aft_dis_speed.split("\\|");
		StringBuffer aft_sb = new StringBuffer();
		for (String dis_speed : aft_dis_speed_other) {
			String[] disspeed = dis_speed.split("\\_");
			aft_sb.append(disspeed[0] + "_" + "-1" + "|");
		}
		String outLine;
		if (flag.equals("0")) {
			outLine = id + "," + lonlat + "," + angle + "," + dir + "," + bef_dis_speed + "," + aft_dis_speed + "," + bef_sb.toString() + "," + aft_sb.toString() + "," + trajMakeTime + "," + total_len + "," + total_len + "," + bef_len + "," + aft_len + "," + bef_len + "," + aft_len + "," + carIdTimeSeq + "," + roadClass;

		} else {
			outLine = id + "," + lonlat + "," + angle + "," + dir + "," + bef_sb.toString() + "," + aft_sb.toString() + "," + bef_dis_speed + "," + aft_dis_speed + "," + trajMakeTime + "," + total_len + "," + total_len + "," + bef_len + "," + aft_len + "," + bef_len + "," + aft_len + "," + carIdTimeSeq + "," + roadClass;
		}
		// 翻转link顺序
		String traffic = getTraffic(outLine);
		// System.out.println("traffic:" + traffic);
		return new HighPrecisionCombineDirTraffic().parseLine(traffic);
	}

	private String getTraffic(String outLine) {
		// TODO Auto-generated method stub
		String[] arr = outLine.split(",");
		String[] befStraight = arr[4].split("\\|");
		String[] befbent = arr[6].split("\\|");
		String[] aftStraight = arr[5].split("\\|");
		String[] aftbent = arr[7].split("\\|");
		String carIdTimeSeq = arr[15];
		String roadClass = arr[16];
		StringBuffer straight_bef = new StringBuffer();

		for (int i = befStraight.length - 1; i >= 0; i--) {
			straight_bef.append(befStraight[i]);
			straight_bef.append("|");
		}

		StringBuffer straight_aft = new StringBuffer();
		for (int i = aftStraight.length - 1; i >= 0; i--) {
			straight_aft.append(aftStraight[i]);
			straight_aft.append("|");
		}
		StringBuffer bent_bef = new StringBuffer();
		for (int j = befbent.length - 1; j >= 0; j--) {
			bent_bef.append(befbent[j]);
			bent_bef.append("|");
		}
		StringBuffer bent_aft = new StringBuffer();
		for (int j = aftbent.length - 1; j >= 0; j--) {
			bent_aft.append(aftbent[j]);
			bent_aft.append("|");
		}
		String outLine1 = arr[0] + "," + arr[1] + "," + arr[2] + "," + arr[3] + "," + straight_bef.toString() + "," + straight_aft.toString() + "," + bent_bef.toString() + "," + bent_aft.toString() + "," + arr[8] + "," + arr[9] + "," + arr[10] + "," + arr[11] + "," + arr[12] + "," + arr[13] + "," + arr[14] + "," + carIdTimeSeq + "," + roadClass;
		return outLine1;
	}

	private HighPrecisionCombineDirTraffic combineTwoLine(ArrayList<HighPrecisionCombineDirTraffic> dataList) {
		// TODO Auto-generated method stub
		// 有的时候由于数据会有延迟，导致kafka接收数据变慢，导致一分钟一个路口有2条以上记录，对于同一道路的数据取距离当前时刻最近的数据
		StringBuffer sb = new StringBuffer();
		HashMap<Integer, HighPrecisionCombineDirTraffic> dataMap = new HashMap<Integer, HighPrecisionCombineDirTraffic>();
		for (HighPrecisionCombineDirTraffic data : dataList) {
			sb.append(data.carIdTimeSeq);
			String[] arrs = data.toString().split(",");
			int flag;
			if (arrs[4].contains("-1")) {
				flag = 0;
			} else {
				flag = 1;
			}
			HighPrecisionCombineDirTraffic d = dataMap.get(flag);
			if (d == null) {
				dataMap.put(flag, data);
			} else {
				long time = Long.parseLong(d.timeStamp_5min);
				long nowTime = Long.parseLong(data.timeStamp_5min);
				if (nowTime > time) {
					dataMap.put(flag, data);
				}
			}
		}

		if (dataMap.size() == 1) {
			return dataMap.entrySet().iterator().next().getValue();
		} else {
			String line1 = dataMap.get(0).toString();
			String line2 = dataMap.get(1).toString();
			String[] arr1 = line1.split(",");
			String[] arr2 = line2.split(",");
			if (arr1[4].contains("-1")) {
				arr1[4] = arr2[4];
				arr1[5] = arr2[5];
				arr1[9] = arr2[9];
				arr1[11] = arr2[11];
				arr1[12] = arr2[12];
			} else {
				arr1[6] = arr2[6];
				arr1[7] = arr2[7];
				arr1[10] = arr2[10];
				arr1[13] = arr2[13];
				arr1[14] = arr2[14];
			}
			String out = "";
			for (int i = 0; i < arr1.length; i++) {
				out += arr1[i];
				out += ",";
			}
			out += sb.toString();
			// System.out.println("out:" + out);
			return new HighPrecisionCombineDirTraffic().parseLine(out);
		}

	}

	public static void main(String[] args) {
		// TODO Auto-generated method stub
		RoadTrafficCombine cm = new RoadTrafficCombine();
		// 直行弯路只有一条的情况，需要补全另一条
		System.out.println("有" + 2 + "条方向数据进行合并!");
		HighPrecisionCombineDirTraffic outLine = cm.combineTwoLine1();
		System.out.println("合并后数据：" + outLine);

	}

	private HighPrecisionCombineDirTraffic combineTwoLine1() {
		// TODO Auto-generated method stub
		StringBuffer sb = new StringBuffer();
		HashMap<Integer, HighPrecisionCombineDirTraffic> dataMap = new HashMap<Integer, HighPrecisionCombineDirTraffic>();

		// a
		String dataStr = "779013,116.37991|39.98289,348.0,11,22_-1|24_-1|,15_-1|53_-1|105_-1|62_-1|,22_26|24_45|,15_45|53_31|105_41|62_46|,0000,279,279,46,233,46,233";
		String carIdTimeSeq = "102539:1467273720#|102539:1467273720#|102539:1467273720#|";
		HighPrecisionCombineDirTraffic data = new HighPrecisionCombineDirTraffic().parseLine(dataStr + "," + carIdTimeSeq + "," + 0);
		sb.append(carIdTimeSeq);
		String[] arrs = dataStr.split(",");
		int flag;
		if (arrs[4].contains("-1")) {
			flag = 0;
		} else {
			flag = 1;
		}
		HighPrecisionCombineDirTraffic d = dataMap.get(flag);
		if (d == null) {
			dataMap.put(flag, data);
		} else {
			long time = Long.parseLong(d.timeStamp_5min);
			long nowTime = Long.parseLong(data.timeStamp_5min);
			if (nowTime > time) {
				dataMap.put(flag, data);
			}
		}

		// b

		String dataStr1 = "779013,116.37991|39.98289,348.0,11,64_55|,67_55|,64_-1|,67_-1|,0000,129,129,64,65,64,65";
		String carIdTimeSeq1 = "3487702:1467273723#|3487702:1467273723#|3487702:1467273723#|";
		HighPrecisionCombineDirTraffic data1 = new HighPrecisionCombineDirTraffic().parseLine(dataStr1 + "," + carIdTimeSeq1 + "," + 0);
		sb.append(carIdTimeSeq1);
		String[] arrs1 = dataStr1.split(",");
		int flag1;
		if (arrs1[4].contains("-1")) {
			flag1 = 0;
		} else {
			flag1 = 1;
		}
		HighPrecisionCombineDirTraffic d1 = dataMap.get(flag1);
		if (d1 == null) {
			dataMap.put(flag1, data1);
		} else {
			long time = Long.parseLong(d1.timeStamp_5min);
			long nowTime = Long.parseLong(data1.timeStamp_5min);
			if (nowTime > time) {
				dataMap.put(flag1, data1);
			}
		}
		if (dataMap.size() == 1) {
			return dataMap.entrySet().iterator().next().getValue();
		} else {
			String line1 = dataMap.get(0).toString();
			String line2 = dataMap.get(1).toString();
			String[] arr1 = line1.split(",");
			String[] arr2 = line2.split(",");
			if (arr1[4].contains("-1")) {
				arr1[4] = arr2[4];
				arr1[5] = arr2[5];
				arr1[9] = arr2[9];
				arr1[11] = arr2[11];
				arr1[12] = arr2[12];
			} else {
				arr1[6] = arr2[6];
				arr1[7] = arr2[7];
				arr1[10] = arr2[10];
				arr1[13] = arr2[13];
				arr1[14] = arr2[14];
			}
			String out = "";
			for (int i = 0; i < arr1.length; i++) {
				out += arr1[i];
				out += ",";
			}
			out += sb.toString();
			// System.out.println("out:" + out);
			return new HighPrecisionCombineDirTraffic().parseLine(out);
		}
	}

}
